package com.atguigu.day03;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.TimeUnit;

public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<SensorReading> streamSource = env.addSource(new SensorSource());
        streamSource.keyBy(x->x.sensorId).process(new KeyedProcessFunction<String, SensorReading, String>() {
            private ValueState<Double> last_value;
            private ValueState<Long> timer;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                last_value= getRuntimeContext().getState(new ValueStateDescriptor<Double>(
                        "last_value", Types.DOUBLE
                ));
                timer= getRuntimeContext().getState(new ValueStateDescriptor<Long>(
                        "timer",Types.LONG
                ));
            }

            @Override
            public void processElement(SensorReading value, Context ctx, Collector<String> out) throws Exception {
                //获取上一次的值并更新本次的值
                Double lastValue = last_value.value();
                last_value.update(value.temperature);
                //获取触发器的值
                Long time = timer.value();

                if (time==null){
                    ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime()+1000);
                    timer.update(ctx.timerService().currentProcessingTime()+1000);
                }else if (lastValue>value.temperature){
                    ctx.timerService().deleteProcessingTimeTimer(time);
                    timer.clear();
                }


            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                out.collect("设备<"+ctx.getCurrentKey()+">连续1秒钟温度持续增加");
                timer.clear();
            }
        }).print();

        env.execute();
    }

    public static class SensorSource implements SourceFunction<SensorReading> {
        private Boolean running = true;
        private final Random random = new Random();
        private final String[] sensorIds = {
                "sensor_0",
                "sensor_1",
                "sensor_2",
                "sensor_3",
                "sensor_4",
                "sensor_5",
                "sensor_6",
                "sensor_7",
                "sensor_8",
                "sensor_9",
        };

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            Double[] doubles = new Double[10];
            for (int i = 0; i < doubles.length; i++) {
                doubles[i]= 65 + 20 * random.nextGaussian();
            }

            while (running) {
                for (int i = 0; i < sensorIds.length; i++) {
                    ctx.collect(SensorReading.of(
                            sensorIds[i],
                            doubles[i] + random.nextGaussian() * 0.5,
                            Calendar.getInstance().getTimeInMillis()
                    ));
                }
                TimeUnit.MILLISECONDS.sleep(10);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

    /**
     * 存储设备ID，温度和时间戳
     */
    public static class SensorReading {
        public String sensorId;
        public Double temperature;
        public Long timestamp;

        public static SensorReading of(String sensorId , Double temperature ,Long timestamp){
            return new SensorReading(sensorId, temperature, timestamp);
        }

        @Override
        public String toString() {
            return "SensorReading{" +
                    "sensorId='" + sensorId + '\'' +
                    ", temperature=" + temperature +
                    ", timestamp=" + new Timestamp(timestamp) +
                    '}';
        }

        public SensorReading(String sensorId, Double temperature, Long timestamp) {
            this.sensorId = sensorId;
            this.temperature = temperature;
            this.timestamp = timestamp;
        }
    }
}
